package mylab.starters.dag.task;


import com.github.rholder.retry.Retryer;
import com.google.common.base.Preconditions;
import com.google.common.collect.Sets;

import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicBoolean;

import mylab.starters.dag.DagContext;
import mylab.starters.dag.DagContextHolder;
import mylab.starters.dag.TaskResult;
import mylab.starters.dag.enums.TaskResultState;
import mylab.starters.dag.graph.Graph;
import mylab.starters.dag.graph.Vertex;
import mylab.utils.common.CastUtil;

/**
 * DAG中的任务节点
 */
public interface Task<V> extends Vertex, TaskSetting<V> {

    TaskExecutor<V> getTaskExecutor();

    /**
     * 节点任务有无执行过
     */
    AtomicBoolean isExecuted();

    default Engine getEngine() {
        Graph graph = getGraph();
        Preconditions.checkState(graph instanceof Engine);
        return CastUtil.cast(graph);
    }

    /**
     * 获取前驱节点的任务结果作为参数
     */
    default Set<TaskResult<?>> getDependentParameters() {
        Set<TaskResult<?>> sets = Sets.newHashSet();
        DagContext dagContext = DagContextHolder.get(getGraph().getId());
        Preconditions.checkNotNull(dagContext, "DAG context is null");
        getDependentVertices().forEach(vertex -> {
            TaskResult<?> taskResult = dagContext.getTaskResult(vertex.getId());
            if (taskResult != null) {
                sets.add(taskResult);
            }
        });
        return sets;
    }

    default void doRun() {
        //通过入度判断前置依赖是否全部完成
        if (getIndegree().get() > 0) {
            return;
        }
        //任务已经执行过了
        if (isExecuted().get()) {
            return;
        }
        if (getEngine().isAborted()) {
            return;
        }

        DagContext context = DagContextHolder.get(getGraph().getId());
        context.getExecutorService().submit(runnable());
    }

    default Runnable runnable() {
        final Retryer<V> retryer = getRetryer();

        return () -> {
            synchronized (Task.this) {
                //全图上下文
                DagContext context = DagContextHolder.get(getGraph().getId());
                TaskResult<V> taskResult = null;
                V result = getDefaultValue();
                boolean isBreak = false;
                try {
                    result = retryer.call(callable());
                    taskResult = new TaskResult<>(getId(), result, TaskResultState.SUCCESS);
                } catch (Exception e) {
                    switch (getTaskFailStrategy()) {
                        case DEFAULT_VALUE_WHEN_FAIL -> {
                            taskResult = new TaskResult<>(getId(), result, TaskResultState.DEFAULT, e);
                        }
                        case BREAK_WHEN_FAIL -> {
                            taskResult = new TaskResult<>(getId(), null, TaskResultState.EXCEPTION, e);
                            isBreak = true;
                        }
                    }
                } finally {
                    //保存任务结果到上下文
                    context.putTaskResult(getId(), taskResult);
                    //修改任务执行标记
                    isExecuted().compareAndSet(false, true);
                    //任务即将完成，本线程即将退出，从上下文删除线程记录
                    context.removeTaskThread(getId());
                    //当前任务完成
                    context.getSyncLatch().countDown();
                    // callback
                    if (getEngine().getEngineCallback() != null) {
                        getEngine().getEngineCallback().onTaskEnd(getId(), taskResult);
                    }
                }

                if (isBreak) {
                    // 任务异常，通知全局中止
                    getEngine().abort(true);
                } else {
                    //修改后置节点入度
                    getAdjacentVertices().forEach(vertex -> {
                        if (vertex instanceof Task<?> nextTask) {
                            nextTask.getIndegree().decrementAndGet();
                        }
                    });
                    //运行后置任务
                    for (Vertex vertex : getAdjacentVertices()) {
                        if (vertex instanceof Task<?> nextTask) {
                            nextTask.doRun();
                        }
                    }
                }
            }
        };
    }

    default Callable<V> callable() {
        return () -> {
            //全图上下文
            DagContext context = DagContextHolder.get(getGraph().getId());
            //记录当前线程
            context.putTaskThread(getId(), Thread.currentThread());
            //获取外部参数
            Set<Object> parameters = getParameters();
            //获取前置任务结果作为参数
            Set<TaskResult<?>> dependentTaskResults = getDependentParameters();

            //callback
            if (getEngine().getEngineCallback() != null) {
                getEngine().getEngineCallback().onTaskBegin(getId(), parameters, dependentTaskResults);
            }

            //执行本节点任务
            return getTaskExecutor().execute(getEngine().getEngineCallback(),
                    parameters, dependentTaskResults);
        };
    }
}
